"""Scheduler runner wiring automation jobs to configurable triggers."""
from __future__ import annotations

import argparse
import asyncio
import signal
import threading
from dataclasses import dataclass, field
from datetime import datetime, time, timedelta, timezone
from threading import Thread
from typing import Callable, Dict, List, Optional, Sequence

try:
    from zoneinfo import ZoneInfo
except ImportError:  # pragma: no cover - Python <3.9 fallback
    from backports.zoneinfo import ZoneInfo  # type: ignore[no-redef]  # pragma: no cover

from app.automation.jobs import (
    run_daily_data_refresh,
    run_preopen_plan,
    run_strategy_review,
)
from app.utils.config import SchedulerSettings, get_config
from app.utils.logging import get_logger

LOGGER = get_logger(__name__)
LOG_EXTRA = {"stage": "scheduler.runner"}


JobCallable = Callable[[], object]


@dataclass
class SchedulerMetrics:
    job_runs: Dict[str, int] = field(default_factory=dict)
    job_failures: Dict[str, int] = field(default_factory=dict)
    last_started: Dict[str, datetime] = field(default_factory=dict)
    last_completed: Dict[str, datetime] = field(default_factory=dict)


_METRICS = SchedulerMetrics()
_METRICS_LOCK = threading.RLock()


def _metrics_record_start(job_name: str) -> None:
    with _METRICS_LOCK:
        _METRICS.last_started[job_name] = datetime.utcnow().replace(tzinfo=timezone.utc)


def _metrics_record_completion(job_name: str, *, success: bool) -> None:
    with _METRICS_LOCK:
        _METRICS.job_runs[job_name] = _METRICS.job_runs.get(job_name, 0) + 1
        _METRICS.last_completed[job_name] = datetime.utcnow().replace(tzinfo=timezone.utc)
        if not success:
            _METRICS.job_failures[job_name] = _METRICS.job_failures.get(job_name, 0) + 1


def get_scheduler_metrics() -> Dict[str, object]:
    """Return a snapshot of scheduler runtime metrics."""

    with _METRICS_LOCK:
        job_names = set(
            list(_METRICS.job_runs.keys())
            + list(_METRICS.job_failures.keys())
            + list(_METRICS.last_started.keys())
            + list(_METRICS.last_completed.keys())
        )
        jobs: List[Dict[str, object]] = []
        for name in sorted(job_names):
            started = _METRICS.last_started.get(name)
            completed = _METRICS.last_completed.get(name)
            jobs.append(
                {
                    "name": name,
                    "runs": _METRICS.job_runs.get(name, 0),
                    "failures": _METRICS.job_failures.get(name, 0),
                    "last_started": started.isoformat() if started else None,
                    "last_completed": completed.isoformat() if completed else None,
                }
            )
        return {
            "generated_at": datetime.utcnow().replace(tzinfo=timezone.utc).isoformat(),
            "jobs": jobs,
        }


def reset_scheduler_metrics() -> None:
    """Reset runtime metrics (useful for tests)."""

    with _METRICS_LOCK:
        _METRICS.job_runs.clear()
        _METRICS.job_failures.clear()
        _METRICS.last_started.clear()
        _METRICS.last_completed.clear()


def _parse_time(value: str, fallback: time) -> time:
    text = (value or "").strip()
    for fmt in ("%H:%M", "%H:%M:%S"):
        try:
            parsed = datetime.strptime(text, fmt)
            return time(parsed.hour, parsed.minute, parsed.second)
        except ValueError:
            continue
    LOGGER.warning("无法解析定时时间 %s，使用默认值 %s", value, fallback, extra=LOG_EXTRA)
    return fallback


@dataclass
class ScheduledJob:
    name: str
    callback: JobCallable
    timezone: ZoneInfo
    time_of_day: time
    enabled: bool = True
    next_run: datetime = field(init=False)

    def __post_init__(self) -> None:
        self.reschedule(reference=datetime.now(self.timezone))

    def reschedule(self, reference: Optional[datetime] = None) -> None:
        if not self.enabled:
            self.next_run = datetime.max.replace(tzinfo=self.timezone)
            return
        now = reference or datetime.now(self.timezone)
        candidate = now.replace(
            hour=self.time_of_day.hour,
            minute=self.time_of_day.minute,
            second=self.time_of_day.second,
            microsecond=0,
        )
        if candidate <= now:
            candidate += timedelta(days=1)
        self.next_run = candidate
        LOGGER.debug(
            "Job %s scheduled for %s",
            self.name,
            self.next_run.isoformat(),
            extra=LOG_EXTRA,
        )

    async def run(self) -> None:
        if not self.enabled:
            return
        LOGGER.info("开始执行定时任务 %s", self.name, extra=LOG_EXTRA)
        _metrics_record_start(self.name)
        try:
            result = await asyncio.to_thread(self.callback)
            if result is not None:
                LOGGER.info("任务 %s 完成 result=%s", self.name, result, extra=LOG_EXTRA)
            else:
                LOGGER.info("任务 %s 完成", self.name, extra=LOG_EXTRA)
            _metrics_record_completion(self.name, success=True)
        except Exception:  # noqa: BLE001
            LOGGER.exception("任务 %s 执行失败", self.name, extra=LOG_EXTRA)
            _metrics_record_completion(self.name, success=False)
        finally:
            self.reschedule()


def _build_scheduled_jobs(settings: SchedulerSettings) -> List[ScheduledJob]:
    if not settings.enabled:
        return []

    tz_name = settings.timezone or "Asia/Shanghai"
    tz = ZoneInfo(tz_name)

    def _wrap(func: Callable[[datetime], object]) -> JobCallable:
        return lambda: func(datetime.now(tz))

    jobs: List[ScheduledJob] = []

    if settings.daily_ingest.enabled:
        ingest_time = _parse_time(settings.daily_ingest.time, time(16, 10))
        jobs.append(
            ScheduledJob(
                name="daily_ingest",
                callback=_wrap(run_daily_data_refresh),
                timezone=tz,
                time_of_day=ingest_time,
                enabled=True,
            )
        )

    if settings.preopen_plan.enabled:
        plan_time = _parse_time(settings.preopen_plan.time, time(8, 45))
        jobs.append(
            ScheduledJob(
                name="preopen_plan",
                callback=_wrap(run_preopen_plan),
                timezone=tz,
                time_of_day=plan_time,
                enabled=True,
            )
        )

    if settings.strategy_review.enabled:
        review_time = _parse_time(settings.strategy_review.time, time(14, 30))
        jobs.append(
            ScheduledJob(
                name="strategy_review",
                callback=_wrap(run_strategy_review),
                timezone=tz,
                time_of_day=review_time,
                enabled=True,
            )
        )

    return jobs


async def run_scheduler(
    jobs: List[ScheduledJob],
    *,
    stop_event: Optional[asyncio.Event] = None,
) -> None:
    if not jobs:
        LOGGER.warning("未找到需要运行的定时任务，退出调度器。", extra=LOG_EXTRA)
        return
    LOGGER.info("定时任务列表：%s", [job.name for job in jobs], extra=LOG_EXTRA)
    external_event = stop_event is not None
    stop_event = stop_event or asyncio.Event()

    def _stop_handler(signum: int, _frame: object) -> None:
        LOGGER.info("接收到终止信号 %s，准备关闭调度器。", signum, extra=LOG_EXTRA)
        stop_event.set()

    loop = asyncio.get_running_loop()
    if threading.current_thread() is threading.main_thread():
        for sig in (signal.SIGTERM, signal.SIGINT):
            try:
                loop.add_signal_handler(sig, _stop_handler, sig, None)
            except NotImplementedError:  # pragma: no cover - Windows fallback
                signal.signal(sig, lambda s, _f: _stop_handler(s, None))

    try:
        while not stop_event.is_set():
            now = datetime.now(jobs[0].timezone)
            next_run = min(job.next_run for job in jobs)
            sleep_seconds = max(0.0, (next_run - now).total_seconds())
            try:
                await asyncio.wait_for(stop_event.wait(), timeout=sleep_seconds)
                break
            except asyncio.TimeoutError:
                pass
            due_jobs = [
                job
                for job in jobs
                if job.next_run <= datetime.now(job.timezone)
            ]
            for job in due_jobs:
                await job.run()
    finally:
        if not external_event and stop_event and not stop_event.is_set():
            stop_event.set()
        LOGGER.info("调度器已停止。", extra=LOG_EXTRA)


def _available_jobs() -> Dict[str, Callable[..., object]]:
    return {
        "daily_ingest": run_daily_data_refresh,
        "preopen_plan": run_preopen_plan,
        "strategy_review": run_strategy_review,
    }


def run_job_once(job_name: str) -> object:
    jobs = _available_jobs()
    if job_name not in jobs:
        raise ValueError(f"未知的任务名称：{job_name}")
    LOGGER.info("手动触发任务 %s", job_name, extra=LOG_EXTRA)
    settings = getattr(get_config(), "scheduler", SchedulerSettings())
    tz_name = settings.timezone or "Asia/Shanghai"
    try:
        tz = ZoneInfo(tz_name)
    except Exception:  # noqa: BLE001
        LOGGER.warning("无法解析时区 %s，回退到 UTC", tz_name, extra=LOG_EXTRA)
        tz = ZoneInfo("UTC")
    return jobs[job_name](datetime.now(tz))


class SchedulerRuntime:
    """Thread-based runtime wrapper to control scheduler lifecycle."""

    def __init__(self) -> None:
        self._thread: Optional[Thread] = None
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._async_stop: Optional[asyncio.Event] = None
        self._jobs: List[ScheduledJob] = []
        self._lock = threading.RLock()

    def is_running(self) -> bool:
        with self._lock:
            return self._thread is not None and self._thread.is_alive()

    def start(self, settings: SchedulerSettings) -> None:
        with self._lock:
            if self.is_running():
                self.stop(wait=True)
            jobs = _build_scheduled_jobs(settings)
            if not jobs:
                raise ValueError("未启用任何定时任务，无法启动调度器。")
            self._jobs = jobs
            self._thread = Thread(target=self._run_thread, args=(jobs,), daemon=True)
            self._thread.start()

    def _run_thread(self, jobs: List[ScheduledJob]) -> None:
        try:
            asyncio.run(self._run_async(jobs))
        finally:
            with self._lock:
                self._loop = None
                self._async_stop = None
                self._thread = None
                self._jobs = []

    async def _run_async(self, jobs: List[ScheduledJob]) -> None:
        self._loop = asyncio.get_running_loop()
        self._async_stop = asyncio.Event()
        await run_scheduler(jobs, stop_event=self._async_stop)

    def stop(self, *, wait: bool = False, timeout: float = 5.0) -> None:
        with self._lock:
            if not self.is_running():
                return
            if self._loop and self._async_stop:
                self._loop.call_soon_threadsafe(self._async_stop.set)
            thread = self._thread
        if wait and thread is not None:
            thread.join(timeout=timeout)

    def status(self) -> Dict[str, object]:
        with self._lock:
            running = self.is_running()
            jobs_snapshot = [
                {
                    "name": job.name,
                    "next_run": job.next_run.isoformat(),
                    "time_of_day": job.time_of_day.strftime("%H:%M"),
                    "timezone": job.timezone.key if hasattr(job.timezone, "key") else str(job.timezone),
                }
                for job in self._jobs
            ]
        return {
            "running": running,
            "jobs": jobs_snapshot,
        }


_RUNTIME: Optional[SchedulerRuntime] = None


def get_scheduler_runtime() -> SchedulerRuntime:
    global _RUNTIME
    if _RUNTIME is None:
        _RUNTIME = SchedulerRuntime()
    return _RUNTIME


def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="后台任务调度器")
    parser.add_argument(
        "--job",
        choices=sorted(_available_jobs().keys()),
        help="指定任务名称并立即执行一次。",
    )
    parser.add_argument(
        "--list",
        action="store_true",
        help="列出当前配置下启用的定时任务并退出。",
    )
    return parser.parse_args(argv)


def main(argv: Optional[Sequence[str]] = None) -> None:
    args = parse_args(argv)
    cfg = get_config()
    settings = getattr(cfg, "scheduler", SchedulerSettings())

    if args.list:
        jobs = _build_scheduled_jobs(settings)
        LOGGER.info(
            "启用的定时任务：%s",
            [
                {
                    "name": job.name,
                    "next_run": job.next_run.isoformat(),
                    "time": job.time_of_day.isoformat(),
                }
                for job in jobs
            ],
            extra=LOG_EXTRA,
        )
        return

    if args.job:
        run_job_once(args.job)
        return

    jobs = _build_scheduled_jobs(settings)
    asyncio.run(run_scheduler(jobs))


if __name__ == "__main__":
    main()
